package com.github.rpc4j.consumer.core;

import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang.StringUtils;

import com.github.rpc4j.common.network.NettyRpcConnection;
import com.github.rpc4j.common.network.NettyRpcConnectionFactory;
import com.github.rpc4j.common.network.RpcConnectionFactory;
import com.github.rpc4j.consumer.clusters.ServiceClustersCall;
import com.github.rpc4j.consumer.clusters.ServiceClustersContext;
import com.github.rpc4j.consumer.exception.Rpc4jConsumerStartException;
import com.github.rpc4j.consumer.zk.ZkUtil;
import com.github.rpc4j.util.ServerUtil;

public class Rpc4jConsumerService {
	private String id;
	private String zkServer;
	private String serviceClustersCallClassName;
	private Rpc4jConsumerConfig  config;
	private Rpc4jConsumerContext context = new Rpc4jConsumerContext();
	private final static AtomicInteger serverSeq = new AtomicInteger(0);
	private RpcConnectionFactory<NettyRpcConnection> rpcConnectionFactory = new NettyRpcConnectionFactory();
	
	public void setId(String id) {
		this.id = id;
	}

	public void setZkServer(String zkServer) {
		this.zkServer = zkServer;
	}
	
	
    

	public void setServiceClustersCallClassName(String serviceClustersCallClassName) {
		this.serviceClustersCallClassName = serviceClustersCallClassName;
	}

	public Rpc4jConsumerContext getContext() {
		return context;
	}

	public void init() {
		try{
			rpcConnectionFactory.init();
			config();
			start();
		}catch(Exception e){
			
		}
	}
	
	private void config(){
		config = new Rpc4jConsumerConfig();
		if(id==null){
		   String ip = ServerUtil.getServerIp();
		   config.setId(ip+"_"+serverSeq.getAndIncrement());
		}else{
			config.setId(id);
		}
		config.setZkServer(zkServer);
		context.setConfig(config);
		context.setRpcConnectionFactory(rpcConnectionFactory);
		initServiceClusters(this.serviceClustersCallClassName);
	}
	
	private void initServiceClusters(String className){
		try{
			Class clzss = Class.forName(className);
			ServiceClustersCall serviceClustersCall = (ServiceClustersCall)clzss.newInstance();
			ServiceClustersContext serviceClustersContext = new ServiceClustersContext(serviceClustersCall);
			context.setServiceClustersContext(serviceClustersContext);
		}catch(Exception e){
			throw new Rpc4jConsumerStartException(e.getMessage());
		}
	}
	
	
	private void start(){
		initZk();
	}
	
	private void initZk() {
		if (StringUtils.isBlank(config.getZkServer()))
			return;
		ZkUtil zkUtil = new ZkUtil(config);
		zkUtil.init();
		context.setZkUtil(zkUtil);
	}


}
